package com.wan.net.rpc.invoker.callback;

import com.wan.core.timer.WTimer;
import com.wan.core.timer.rule.WRules;
import com.wan.net.exception.RpcInvokeException;
import com.wan.net.rpc.message.RpcResponse;
import com.wan.net.rpc.util.ResultCode;
import com.wan.util.Constants;
import com.wan.util.cache.Caches;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;

/**
 * 消息回调管理
 * 管理从服务器响应的消息，需要定时清理掉失效消息
 * @author wwh
 * @date 2022/6/22 17:07
 */
public class CallbackManager {
    private static final Logger logger = LoggerFactory.getLogger(CallbackManager.class);
    private static final Caches<Integer,RpcFuture<RpcResponse>> RPC_FUTURE_CACHES;

    static {
        RPC_FUTURE_CACHES = Caches.<Integer,RpcFuture<RpcResponse>>builder()
                .setSize(2000)
                //过期时间2分钟
                .setTimeout(Constants.MILLS_OF_MINUTE * 2)
                //.setPeriod(Constants.MILLS_OF_MINUTE * 2) //不使用额外的定时器
                .build();
        //使用时间轮管理,5分钟后每2分钟清理一次
        WTimer.get().repeat(node->RPC_FUTURE_CACHES.prune(),
                WRules.MINUTE_REPEAT,2, LocalDateTime.now().plusMinutes(5));
    }

    public static void addFuture(int messageId,RpcFuture<RpcResponse> future){
        RPC_FUTURE_CACHES.put(messageId,future);
    }

    public static void callback(int messageId,RpcResponse response){
        ResultCode resultCode = ResultCode.of(response.getCode());
        RuntimeException runtimeException = null;
        if(resultCode != ResultCode.SUCCESS){
            runtimeException = new RpcInvokeException("unknown_msgId:" + messageId,
                    resultCode == null ? "未知错误码": resultCode.getExplain());
        }
        RpcFuture<RpcResponse> rpcFuture = RPC_FUTURE_CACHES.get(messageId);
        if(rpcFuture == null){
            logger.error("[rpc]消息回调失效，msgId:{},result:{}",messageId,response,runtimeException);
            return;
        }
        if(runtimeException != null){
            rpcFuture.error(runtimeException);
        }else{
            rpcFuture.complete(response);
        }
        RPC_FUTURE_CACHES.remove(messageId);
    }

}
